package com.binance.notification.web.kafka;

import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.binance.notification.web.service.IAlertPriceService;
import com.binance.notification.web.service.IAlertTradeService;

import lombok.extern.slf4j.Slf4j;

@EnableBinding(IChannel.class)
@Slf4j
public class Consumer {

    @Autowired
    private IAlertPriceService alertPriceService;

    @Autowired
    private IAlertTradeService alertTradeService;

    @Value("${jpush.alert.price.enabled}")
    private boolean alertPriceEnabled;

    @Value("${jpush.alert.trade.enabled}")
    private boolean alertTradeEnabled;

    @StreamListener(IChannel.ALERT_PRICE_PAIR_KAFKA_INPUT_CHANNEL)
    public void receivePriceReminderPair(Object payload) {
        if (!alertPriceEnabled || payload == null) {
            return;
        }
        String payloadString = this.parsePayload(payload);
        log.info("收到消息：{}", payloadString);
        String[] array = StringUtils.split(payloadString, "|");
        this.alertPriceService.putPair(array[0], Double.parseDouble(array[1]));
    }

    @StreamListener(IChannel.ALERT_PRICE_KAFKA_INPUT_CHANNEL)
    public void receivePriceReminderStr(Object payload) {
        if (!alertPriceEnabled || payload == null) {
            return;
        }
        String payloadString = this.parsePayload(payload);
        log.info("收到消息：{}", payloadString);
        String[] array = StringUtils.split(payloadString, ";");
        this.alertPriceService.putAlertPrice(array[0], Double.parseDouble(array[1]));
    }

    /**
     * 监听交易提醒通道
     *
     * @param payload
     */
    @StreamListener(IChannel.ALERT_TRADE_KAFKA_INPUT_CHANNEL)
    public void receiveDealReminder(Object payload) {
        if (!alertTradeEnabled || payload == null) {
            return;
        }
        String message = null;
        if (payload instanceof String) {
            message = (String) payload;
        } else {
            message = new String((byte[]) payload);
        }
        JSONObject object = JSON.parseObject(message);
        JSONArray array = object.getJSONArray("events");
        for (int i = 0; i < array.size(); i++) {
            JSONObject event = array.getJSONObject(i);
            // 属于交易的数据
            if (StringUtils.equals(event.getString("eventType"), "executionReport")
                    && StringUtils.equals(event.getString("executionType"), "TRADE")) {
                String userId = event.getString("externalId");
                String pair = event.getString("symbol");
                // 0 买，1 卖
                String type = StringUtils.equals("BUY", event.getString("side")) ? "0" : "1";
                String price = event.getString("price");
                String amount = event.getString("qty");
                // 0 部分成交，1 完全成交
                String status = StringUtils.equals(event.getString("orderStatus"), "PARTIALLY_FILLED") ? "0" : "1";
                String date = event.getString("time");
                String orderId = event.getString("orderId");
                String payloadString = userId + "," + pair + "," + type + "," + price + "," + amount + "," + status
                        + "," + date + "," + orderId;
                log.info("收到消息：{}", payloadString);
                this.alertTradeService.pushForAlert(payloadString);
            }
        }
    }

    private String parsePayload(Object payload) {
        String payloadString = payload.toString();
        if (payloadString.startsWith("\"") && payloadString.endsWith("\"")) {
            payloadString = StringUtils.substring(payloadString, 1, payloadString.length() - 1);
        }
        return payloadString;
    }
}
